package pipeline;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * Created by Administrator on 2017/3/17.
 */
public abstract class AbstractParallelPipe<IN,OUT,V> extends AbstractPipe<IN,OUT> {

    private final ExecutorService executorService;

    public AbstractParallelPipe(BlockingQueue<IN> queue,ExecutorService executorService) {
        super();
        this.executorService = executorService;
    }

    /**
     * 留给子类去实现，用于根据指定的输入元素构造一组子任务。
     * @param input
     * @return
     * @throws Exception
     */
    protected abstract List<Callable<V>> buildTasks(IN input) throws Exception;

    /**
     * 留给子类去实现，用于对各个子任务的处理结果进行合并。
     * @param subTaskResults
     * @return
     * @throws Exception
     */
    protected abstract OUT combineResults(List<Future<V>> subTaskResults) throws Exception;

    /**
     * 以并行的方式 执行一组子任务
     * @param tasks
     * @return
     * @throws Exception
     */
    protected List<Future<V>> invokeParallel (List<Callable<V>> tasks) throws Exception{
        return executorService.invokeAll(tasks);
    }

    @Override
    protected OUT doProcess(final IN input) throws PipeException {
        OUT out = null ;
        try{
            out = combineResults(invokeParallel(buildTasks(input)));
        }catch (Exception e){
            throw new PipeException(this,input,"Task failed ",e);
        }
        return out;
    }

}
